#include <iostream>
#include <sstream>

#include <gtest/gtest.h>

#include "Producer.h"
#include "Consumer.h"

#include <thread>
#include <chrono>

TEST(testProducer, use){
    { 
        auto producer = streamdb::makeProducer("./streamdb", "topic");
        auto s = producer->append("1:hello1");
        ASSERT_TRUE(s.isOk());
        s = producer->append("2:hello2");
        ASSERT_TRUE(s.isOk());

    }
    {
        auto producer = streamdb::makeProducer("./streamdb", "topic1");
        auto s = producer->append("1:helll1");
        ASSERT_TRUE(s.isOk());
        s = producer->append("2:helll2");
        ASSERT_TRUE(s.isOk());
    }
}

TEST(ConsumerTest, use){
    auto consumer = streamdb::makeConsumer("./streamdb");
    ASSERT_TRUE(consumer != nullptr);

    auto s= consumer->subscribe("topic", 0);
    ASSERT_TRUE(s.isOk());

    s= consumer->subscribe("topic1", 0);
    ASSERT_TRUE(s.isOk());

    auto iter = consumer->all(); //read all msg already in filesystem
    if(iter != nullptr){
        while(iter->hasNext()){
            auto str = iter->next();
            //TODO: std::cout << (str == nullptr ? "nullstring" : *str)<< std::endl;
        }
    }

    consumer->close();
}

TEST(ConsumerTest, poll){
    
    auto consumer = streamdb::makeConsumer("./streamdb");
    assert(consumer != nullptr);
    auto s = consumer->subscribe("topic", 0);
    assert(s.isOk());

    //TODO: while(true){  
        auto iter = consumer->poll(std::chrono::milliseconds(500)); //consumer save the curPos in meta file
        if(iter != nullptr){
            while(iter->hasNext()){
                auto str = iter->next();
                //TODO: std::cout << (str == nullptr ? "nullstring" : *str)<< std::endl;
            }
        }
    //}
    consumer->close();
}

TEST(ProducerConsumerTest, use){
    std::thread threadProducer([](){
        { 
            auto producer = streamdb::makeProducer("./streamdb", "topic");
            int count = 0;
            for(int j = 0 ; j < 10; ++j){
                for(int i = 1; i <= 10; ++i){
                    count ++;
                    //std::cout << "count:" << count<< std::endl;
                    std::ostringstream oss;
                    oss << count << ":hello" << count;
                    producer->append(oss.str());
                }

                std::this_thread::sleep_for(std::chrono::milliseconds(20));
            }
        }
    });

    std::thread threadConsumer([](){
        auto consumer = streamdb::makeConsumer("./streamdb");
        assert(consumer != nullptr);

        auto s= consumer->subscribe("topic", 0);
        assert(s.isOk());

        for(int j = 0 ; j < 10; ++j){
            //std::cout << "jjjjjjjjjj...." << std::endl;
            auto iter = consumer->all(); //read all msg already in filesystem
            if(iter != nullptr){
                while(iter->hasNext()){
                    auto str = iter->next();
                    //std::cout << (str == nullptr ? "nullstring" : *str)<< std::endl;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
        }

        consumer->close();
    });

    threadProducer.join();
    threadConsumer.join();
}


